package com.atguigu.gmall.cdc;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

/**
 * Author: Felix
 * Date: 2022/1/15
 * Desc: 使用FlinkCDC读取mysql数据---自定义反序列化方式
 */
public class FlinkCDC03_CustomSchema {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
            .hostname("hadoop202")
            .port(3306)
            .databaseList("gmall0722_realtime")
            .tableList("gmall0722_realtime.t_user")
            .username("root")
            .password("123456")
            .startupOptions(StartupOptions.initial())
            .deserializer(new MyDeserializationSchema())
            .build();

        env.addSource(sourceFunction).print(">>>");

        env.execute("Print MySQL Snapshot + Binlog");
    }
}

class MyDeserializationSchema implements DebeziumDeserializationSchema<String> {

    /*
    {
	    "database":"gmall0722",
	    "table":"comment_info",
	    "type":"insert",
	    "data":{"id":1481820042930405377,"user_id":68,"nick_name":null}
    }
    * */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> out) throws Exception {
        JSONObject resJsonObj = new JSONObject();
        Struct valueStruct = (Struct)sourceRecord.value();
        Struct sourceStruct = valueStruct.getStruct("source");

        //获取数据库名
        String database = sourceStruct.getString("db");
        //获取表名
        String table = sourceStruct.getString("table");
        //获取操作类型
        String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();

        if("create".equals(type)){
            type = "insert";
        }
        //定义一个json对象，用于封装影响的记录
        Struct afterStruct = valueStruct.getStruct("after");
        JSONObject dataJsonObj = new JSONObject();
        if(afterStruct != null){
            List<Field> fieldList = afterStruct.schema().fields();
            for (Field field : fieldList) {
                String fieldName = field.name();
                Object fieldValue = afterStruct.get(field);
                dataJsonObj.put(fieldName,fieldValue);
            }
        }


        //封装返回的json对象
        resJsonObj.put("database",database);
        resJsonObj.put("table",table);
        resJsonObj.put("type",type);
        resJsonObj.put("data",dataJsonObj);

        out.collect(resJsonObj.toJSONString());

    }

    @Override
    public TypeInformation<String> getProducedType() {
        return TypeInformation.of(String.class);
    }
}
